package bma.common.netty.thrift.dual;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;

import bma.common.langutil.ai.AIUtil;
import bma.common.langutil.ai.common.AITimeoutException;
import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.ai.stack.AIStackROOT;
import bma.common.langutil.concurrent.ProcessTimerTaskAbstract;
import bma.common.langutil.concurrent.TimerManager;
import bma.common.langutil.core.DateTimeUtil;
import bma.common.langutil.core.ExceptionUtil;
import bma.common.langutil.io.HostPort;
import bma.common.netty.client.NettyClient;
import bma.common.netty.client.NettyClientBootstrap;
import bma.common.netty.handler.ChannelHandlerExceptionClose;
import bma.common.netty.handler.ChannelHandlerFramed;
import bma.common.netty.handler.ChannelHandlerPlaceholder;
import bma.common.netty.thrift.core.NettyThriftConfig;
import bma.common.netty.thrift.dual.impl.NTSDispatcherCommon;
import bma.common.netty.thrift.dual.impl.NTSModuleProcessor;

public class NettyThriftDualClient {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NettyThriftDualClient.class);

	protected NettyThriftConfig config;
	protected NTSProcessor processor;
	protected List<HostPort> serverHosts = new ArrayList<HostPort>();
	protected NettyClientBootstrap bootstrap;
	protected long delayConnectTime = 1000;

	// runtime
	protected NettyClient nettyClient;
	protected int hostIndex = 0;
	protected static final int STOP = -1;
	protected static final int DISCONNECT = 0;
	protected static final int CONNECTING = 1;
	protected static final int CONNECTED = 2;
	protected AtomicInteger status = new AtomicInteger(DISCONNECT);
	protected boolean connectErrorLog = false;
	protected List<AIStack<NTSClient>> await = new ArrayList<AIStack<NTSClient>>();

	public NettyThriftDualClient() {
		super();
	}

	public NTSProcessor getProcessor() {
		return processor;
	}

	public void setProcessor(NTSProcessor p) {
		this.processor = p;
	}

	public void setServices(List<NTSModuleProcessor> services) {
		if (this.processor != null) {
			throw new IllegalArgumentException("processor already set");
		}
		NTSDispatcherCommon dis = new NTSDispatcherCommon();
		dis.setServices(services);
		this.processor = dis;
	}

	public NettyThriftConfig getConfig() {
		return config;
	}

	public void setConfig(NettyThriftConfig config) {
		this.config = config;
	}

	public HostPort getServerHost() {
		return hostIndex < serverHosts.size() ? serverHosts.get(hostIndex)
				: null;
	}

	public void setServerHost(HostPort serverHost) {
		this.serverHosts.clear();
		this.serverHosts.add(serverHost);
	}

	public void setServer(String s) {
		setServerHost(HostPort.fromHostString(s, 7070));
	}

	public void setServerHosts(List<HostPort> shlist) {
		this.serverHosts.clear();
		this.serverHosts.addAll(shlist);
	}

	public void setServers(List<String> slist) {
		this.serverHosts.clear();
		for (String s : slist) {
			this.serverHosts.add(HostPort.fromHostString(s, 7070));
		}
	}

	public NettyClientBootstrap getBootstrap() {
		return bootstrap;
	}

	public void setBootstrap(NettyClientBootstrap bootstrap) {
		this.bootstrap = bootstrap;
	}

	public long getDelayConnectTime() {
		return delayConnectTime;
	}

	public void setDelayConnectTime(long delayStartTime) {
		this.delayConnectTime = delayStartTime;
	}

	public void delayStart() {
		delayConnect();
	}

	public void nextHost() {
		hostIndex++;
		if (hostIndex >= serverHosts.size())
			hostIndex = 0;
	}

	protected void reconnect() {
		nextHost();
		delayConnect();
	}

	protected void delayConnect() {
		if (status.get() != DISCONNECT) {
			if (log.isDebugEnabled()) {
				log.debug("client {}, skip reconnect",
						statusTitle(status.get()));
			}
			return;
		}
		if (log.isDebugEnabled()) {
			log.debug("wait {} to connect {}",
					DateTimeUtil.formatPeriod(delayConnectTime),
					getServerHost());
		}
		ProcessTimerTaskAbstract task = new ProcessTimerTaskAbstract(
				System.currentTimeMillis() + delayConnectTime, false) {

			@Override
			public void run() {
				try {
					connect();
				} catch (Exception e) {
					log.warn("[" + getServerHost()
							+ "] connect protobuf client fail", e);
				}
			}

		};
		TimerManager.checkInstance(config.getTimer()).postTimerTask(task);
	}

	public void start() {
		connect();
	}

	public void close() {
		stop();
	}

	public void stop() {
		int s = status.getAndSet(STOP);
		if (s == CONNECTED) {
			if (nettyClient != null) {
				NettyClient nc = nettyClient;
				synchronized (this) {
					nettyClient = null;
				}
				nc.close();
			}
		}
		List<AIStack<NTSClient>> wlist;
		synchronized (this) {
			wlist = new ArrayList<AIStack<NTSClient>>(await);
			await.clear();
		}
		if (!wlist.isEmpty()) {
			Throwable t = new IOException("closed");
			for (AIStack<NTSClient> stack : wlist) {
				AIUtil.safeFailure(stack, t);
			}
		}
	}

	public boolean getClient(final AIStack<NTSClient> stack, int timeout) {
		int s = status.get();
		if (s == STOP) {
			Throwable t = new IOException("closed");
			return AIUtil.safeFailure(stack, t);
		}
		NTSClient r = null;
		synchronized (this) {
			if (nettyClient != null) {
				r = createClient();
			} else {
				if (timeout != 0) {
					await.add(stack);
					if (timeout > 0) {
						ProcessTimerTaskAbstract task = new ProcessTimerTaskAbstract(
								System.currentTimeMillis() + timeout, true) {

							@Override
							public void run() {
								synchronized (this) {
									if (!await.remove(stack))
										return;
								}
								config.getExecutor().execute(new Runnable() {
									@Override
									public void run() {
										Throwable t = new AITimeoutException();
										AIUtil.safeFailure(stack, t);
									}
								});
							}
						};
						TimerManager.checkInstance(config.getTimer())
								.postTimerTask(task);
					}
					return false;
				}
			}
		}
		return AIUtil.safeSuccess(stack, r);
	}

	public static String statusTitle(int v) {
		switch (v) {
		case STOP:
			return "STOP";
		case DISCONNECT:
			return "DISCONNECT";
		case CONNECTING:
			return "CONNECTING";
		case CONNECTED:
			return "CONNECTED";
		default:
			return "Unknow(" + v + ")";
		}
	}

	protected NTSClient createClient() {
		NettyClient nc = nettyClient;
		if (nc == null)
			return null;
		if (!nc.isOpen()) {
			return null;
		}
		Channel ch = nc.getChannel();
		if (ch == null)
			return null;
		return NCHThriftService.getNTSClientFromChannel(ch);		
	}

	public void connect() {
		boolean check = status.compareAndSet(DISCONNECT, CONNECTING);
		if (!check) {
			if (log.isDebugEnabled()) {
				log.debug("client {}, skip connect", statusTitle(status.get()));
			}
			return;
		}

		HostPort serverHost = getServerHost();
		if (serverHost == null) {
			if (log.isWarnEnabled()) {
				log.warn("server host is null");
			}
			return;
		}
		SocketAddress remoteAddress = serverHost.createInetSocketAddress();
		AIStack<NettyClient> stack = new AIStackROOT<NettyClient>() {

			@Override
			public boolean end(final NettyClient result, Throwable t) {
				if (t != null) {
					if (!connectErrorLog) {
						connectErrorLog = true;
						log.info("connect " + getServerHost() + " fail", t);
					} else {
						if (log.isDebugEnabled()) {
							log.debug("connect " + getServerHost() + " fail", t);
						}
					}
					status.compareAndSet(CONNECTING, DISCONNECT);
					reconnect();
					return true;
				}
				connectErrorLog = false;
				boolean check = status.compareAndSet(CONNECTING, CONNECTED);
				if (!check) {
					if (log.isDebugEnabled()) {
						log.debug("client {} after connected, close it",
								statusTitle(status.get()));
					}
					result.close();
					status.compareAndSet(CONNECTED, DISCONNECT);
					reconnect();
					return true;
				}
				result.getCloseFuture().addListener(
						new ChannelFutureListener() {

							@Override
							public void operationComplete(ChannelFuture future)
									throws Exception {
								status.compareAndSet(CONNECTED, DISCONNECT);
								reconnect();
							}
						});
				synchronized (this) {
					nettyClient = result;
				}
				if (log.isDebugEnabled()) {
					log.debug("connected {}", getServerHost());
				}
				AIStackROOT<Boolean> authStack = new AIStackROOT<Boolean>() {

					@Override
					public boolean end(Boolean authed, Throwable t) {
						if (t != null || !authed) {
							if (log.isDebugEnabled()) {
								log.debug("client auth fail ,close it {}/{}",
										result, t);
							}
							result.close();
							status.compareAndSet(CONNECTED, DISCONNECT);
							reconnect();
							return true;
						}

						final List<AIStack<NTSClient>> qlist = new ArrayList<AIStack<NTSClient>>(
								await.size());
						synchronized (this) {
							qlist.addAll(await);
							await.clear();
						}
						config.getExecutor().execute(new Runnable() {

							@Override
							public void run() {
								NTSClient c = createClient();
								for (AIStack<NTSClient> q : qlist) {
									AIUtil.safeSuccess(q, c);
								}
							}
						});
						return true;

					}
				};
				// if (auth != null) {
				// return auth.doAuth(authStack, createClient());
				// } else {
				// return authStack.success(true);
				// }
				return authStack.success(true);
			}

		};

		if (log.isDebugEnabled()) {
			log.debug("connecting {}", serverHost);
		}
		this.bootstrap.connect(stack, remoteAddress, null,
				new ChannelPipelineFactory() {

					@Override
					public ChannelPipeline getPipeline() throws Exception {
						ChannelPipeline pipeline = Channels.pipeline();
						pipeline.addLast("framed", new ChannelHandlerFramed(
								config.getMaxLength()));
						config.sureTimeout();
						pipeline.addLast("service", new NCHThriftService(
								config, processor));
						pipeline.addLast(NettyClient.NETTY_CLIENT_INSTANCE,
								new ChannelHandlerPlaceholder());
						pipeline.addLast("error",
								new ChannelHandlerExceptionClose() {
									@Override
									protected void logException(
											ChannelHandlerContext ctx,
											ExceptionEvent e) {
										if (log.isWarnEnabled()) {
											Throwable t = ExceptionUtil.cause(e
													.getCause());
											log.warn(
													"{} error {}/{}",
													new Object[] {
															ctx.getChannel()
																	.getRemoteAddress(),
															t.getClass()
																	.getName(),
															t.getMessage() });
										}
									}
								});
						return pipeline;
					}
				});
	}

}
